package com.flying.rp.chapter3;

import com.flying.rp.chapter3.model.NewsLetter;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.Objects;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicInteger;

public class NewsServiceSubscriber implements Subscriber<NewsLetter> {
    private final Queue<NewsLetter> mailbox = new ConcurrentLinkedDeque<>();
    private final int take;
    private final AtomicInteger remaining = new AtomicInteger();
    Subscription subscription;

    public NewsServiceSubscriber(int take) {
        this.take = take;
        this.remaining.set(take);
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        if (Objects.isNull(this.subscription)) {
            this.subscription = subscription;
            this.subscription.request(take);
        } else {
            subscription.cancel();
        }

    }

    @Override
    public void onNext(NewsLetter newsLetter) {
        Objects.requireNonNull(newsLetter);
        mailbox.offer(newsLetter);
    }

    @Override
    public void onError(Throwable throwable) {
        Objects.requireNonNull(throwable);
        if (throwable instanceof ResubscribableErrorLettter) {
            subscription = null;
            ((ResubscribableErrorLettter) throwable).resubscribe(this);
        }
    }

    @Override
    public void onComplete() {
        subscription = null;
    }

    public Optional<NewsLetter> eventuallyReadDigest() {
        NewsLetter letter = mailbox.poll();
        if (Objects.nonNull(letter)) {
            if (remaining.decrementAndGet() == 0) {
                this.subscription.request(take);
                remaining.set(take);
            }
            return Optional.of(letter);
        }
        return Optional.empty();
    }
}
